package net.sourceforge.peertwitter.mbpast;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Observable;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;

import net.sourceforge.peertwitter.core.Profile;
import net.sourceforge.peertwitter.core.messaging.TweetMessage;
import net.sourceforge.peertwitter.mbpast.messaging.MBPastMessage;
import net.sourceforge.peertwitter.mbpast.messaging.ReplicaIdsMessage;
import net.sourceforge.peertwitter.mbpast.messaging.ReplicaRequest;
import net.sourceforge.peertwitter.mbpast.messaging.ReplicaRequestAck;
import net.sourceforge.peertwitter.mbpast.messaging.ReplicaTimestampRequest;
import net.sourceforge.peertwitter.mbpast.messaging.ReplicaTimestampRequestAck;
import net.sourceforge.peertwitter.mbpast.messaging.StoreSwarmMessage;
import net.sourceforge.peertwitter.scamp.Swarm;
import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.p2p.commonapi.Application;
import rice.p2p.commonapi.DeliveryNotification;
import rice.p2p.commonapi.Endpoint;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Id.Distance;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.MessageReceipt;
import rice.p2p.commonapi.Node;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.commonapi.NodeHandleSet;
import rice.p2p.commonapi.RouteMessage;
import rice.selector.TimerTask;

public class MBPastImpl extends Observable implements Application,
		ReplicaManager, MBPast {
	private final Node node;
	private final Endpoint endpoint;
	private final Environment env;
	private MBStorage mbs;
	private Set<Id> ignoredIds;
	private Set<Id> knownIds;
	private ReplicaDownloader downloader;
	private Logger logger;
	private boolean nodeIdsModified;
	private Object idsSync = new Object();

	// XXX create a request factory..

	public MBPastImpl(Environment env, Node node, Profile myProfile,Endpoint endpoint) {
		if (node == null || node.getId() == null) {
			throw new RuntimeException("The node is not ready..");
		}
		this.logger = env.getLogManager().getLogger(MBPastImpl.class,
				node.getId().toString());
		this.env = env;
		this.node = node;
		this.endpoint = endpoint;
		this.mbs = new MBStorageImpl(this.node, env);
		this.ignoredIds = new HashSet<Id>();
		mbs.storeProfile(myProfile);
		knownIds = new HashSet<Id>();
		downloader = new ReplicaDownloaderImpl();
		// new Thread(downloader).start();
		synchronized (idsSync) {
			nodeIdsModified = true;
			knownIds.add(node.getId());
		}
		distributeMyReplicaId(); // XXX: a better control may be done
	}

	private class ReplicaDownloaderImpl implements Runnable, ReplicaDownloader {
		private Map<Id, ReplicaSelector> selectors = new HashMap<Id, MBPastImpl.ReplicaSelector>();
		private BlockingQueue<Id> toDownload = new PriorityBlockingQueue<Id>();
		private Set<Id> downloading = new HashSet<Id>();
		private Map<Id, NodeHandle> profileSources = new HashMap<Id, NodeHandle>();
		private Map<Id, NodeHandle> updateSources = new HashMap<Id, NodeHandle>();

		public ReplicaDownloaderImpl() {
			Executors.newSingleThreadExecutor().execute(this);
		}

		@Override
		public void run() {
			while (true) {
				try {
					downloadReplication(toDownload.take());
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

		private void downloadReplication(final Id replication) {
			ReplicaSelector rs = createSelector(replication);

			if (rs.getSources().isEmpty()) {
				logger.log("no sources for: " + replication
						+ "enqueuing for later..");
				Executors.newSingleThreadExecutor().execute(new Runnable() {

					@Override
					public void run() {
						try {
							env.getTimeSource().sleep(5 * 1000);
							logger.log("retrying download of: " + replication);
							toDownload.put(replication);
						} catch (InterruptedException e) {
							logger.logException(
									"Interrupted while retrying replica download",
									e);
						}
					}
				});
			} else {
				logger.log("sources found for: " + replication);
				PriorityQueue<ReplicaSource> profiles = new PriorityQueue<ReplicaSource>(
						rs.getSources().size(),
						new Comparator<ReplicaSource>() {
							@Override
							public int compare(ReplicaSource o1,
									ReplicaSource o2) { // descending order (o2,
														// o1)
								return new Long(o2.getTimestamp().getProfileTimestamp()).compareTo(new Long(o1.getTimestamp().getProfileTimestamp()));
							}
						});

				PriorityQueue<ReplicaSource> updates = new PriorityQueue<ReplicaSource>(
						rs.getSources().size(),
						new Comparator<ReplicaSource>() {
							@Override
							public int compare(ReplicaSource o1,
									ReplicaSource o2) { // descending order (o2,
														// o1)
								return new Long(o2.getTimestamp().getLastUpdateTimestamp()).compareTo(new Long(o1.getTimestamp().getLastUpdateTimestamp()));

							}
						});

				profiles.addAll(rs.getSources());
				updates.addAll(rs.getSources());

				NodeHandle profileSource = profiles.remove().getNode();
				profileSources.put(replication, profileSource);
				NodeHandle updateSource = updates.remove().getNode();
				updateSources.put(replication, updateSource);

				requestReplication(profileSource, replication);

				if (!profileSource.equals(updateSource)) {
					requestReplication(updateSource, replication);
				}

				Executors.newSingleThreadExecutor().execute(new Runnable() {

					@Override
					public void run() {
						try {
							env.getTimeSource().sleep(5 * 1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						if (isDownloading(replication)) {
							scheduleDownload(replication); // to recover..
							throw new RuntimeException("Problem downloading..");
						}

					}
				});
			}
			disposeSelector(replication);
		}

		private synchronized ReplicaSelector createSelector(Id replication) {
			ReplicaSelector newSelector = new ReplicaSelector(replication);
			selectors.put(replication, newSelector);
			return newSelector;
		}

		private void disposeSelector(Id replication) {
			selectors.remove(replication);
		}

		@Override
		public boolean isDownloading(Id replicaId) {
			return downloading.contains(replicaId);
		}

		private void requestReplication(NodeHandle source, Id replication) {
			endpoint.route(source.getId(),
					new ReplicaRequest(node.getLocalNodeHandle(), replication),
					source);
		}

		@Override
		public synchronized void receivedReplication(ReplicaRequestAck message) {
			if (!profileSources.containsKey(message.getReplication().getId()) && !updateSources.containsKey(message.getReplication().getId())) {
				logger.logException("non-expected replication received " + message,
						new RuntimeException());
			} else {
				NodeHandle profile = profileSources.get(message
						.getReplication().getId());
				NodeHandle update = updateSources.get(message.getReplication()
						.getId());

				if (message.getSender().equals(profile)) {
					mbs.storeProfile(message.getReplication().getProfile());
					mbs.storeSwarmView(message.getReplication().getSwarm());
					profileSources.remove(message.getReplication().getId());
				}
				if (message.getSender().equals(update)) {
					mbs.storeUpdates(message.getReplication().getId(), message
							.getReplication().getLastUpdates());
					mbs.storeSwarmView(message.getReplication().getSwarm());
					updateSources.remove(message.getReplication().getId());
				}

				if (!profileSources.containsKey(message.getReplication()
						.getId())
						&& !updateSources.containsKey(message.getReplication()
								.getId())) {
					downloading.remove(message.getReplication().getId());
				}
				mbs.storeReplication(message.getReplication());
			}
		}

		@Override
		public synchronized void scheduleDownload(Id replicaId) {
			downloading.add(replicaId);
			try {
				toDownload.put(replicaId);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		@Override
		public synchronized void receivedTimestamp(
				ReplicaTimestampRequestAck message) {
			if (!selectors.containsKey(message.getReplicationId())) {
				logger.log("i dont have a requester");
			} else {
				selectors.get(message.getReplicationId()).receiveTimestamp(
						message);
			}
		}

	}

	private class ReplicaSelector {
		private static final int TIMEOUT = 5;
		private final List<ReplicaSource> sources;
		private final NodeHandleSet replicaSet;
		private final CountDownLatch countDown;
		private final TimestampRequester requester;

		public ReplicaSelector(Id replicaId) {
			replicaSet = endpoint.replicaSet(replicaId, MBPast.alfa);
			replicaSet.removeHandle(endpoint.getId()); // remove myself
			countDown = new CountDownLatch(replicaSet.size());
			requester = new TimestampRequester(endpoint, replicaId, replicaSet);
			sources = new ArrayList<ReplicaSource>();

			Executors.newSingleThreadScheduledExecutor().execute(requester);
			scheduleUnblockingTimer();
		}

		private void scheduleUnblockingTimer() {
			// we can't use countdown latch because of the simulation timeunit
			env.getSelectorManager().schedule(new TimerTask() {
				@Override
				public void run() {
					while (countDown.getCount() > 0)
						countDown.countDown();
				}
			}, TIMEOUT * 1000);
		}

		public synchronized void receiveTimestamp(ReplicaTimestampRequestAck rta) {
			sources.add(new ReplicaSource(rta.getSender(), rta.getTimestamp(),
					rta.getReplicationId()));
			countDown.countDown();
		}

		public /*synchronized*/ List<ReplicaSource> getSources() {
			try {
				countDown.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return this.sources;
		}
	}

	private void distributeMyReplicaId() {
		NodeHandleSet replicaSet = endpoint.replicaSet(this.node.getId(), alfa);
		replicaSet.removeHandle(this.node.getId());

		for (int current = 0; current < replicaSet.size(); current++) {
			NodeHandle currentNode = replicaSet.getHandle(current);
			endpoint.route(
					currentNode.getId(),
					new ReplicaIdsMessage(this.node.getLocalNodeHandle(), mbs
							.getIds(), new GregorianCalendar().getTime()),
					currentNode);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see br.ufsc.das.mbpast.ReplicaManager#isReplica(rice.p2p.commonapi.Id)
	 */
	@Override
	public boolean isReplica(Id content) {
		return endpoint.range(node.getLocalNodeHandle(), alfa, content, true)
				.containsId(content);
	}

	public boolean isManager(Id content) {
//		boolean result = endpoint.range(node.getLocalNodeHandle(), 0, content,true).containsId(content);
//		IdRange range = endpoint.range(node.getLocalNodeHandle(), 0, content);
//		System.out.println(this.node.getId() + " manager " + content + ":" + result + " " + range);
		return content.equals(node.getId()) || endpoint.replicaSet(content, 1).memberHandle(node.getId());
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see br.ufsc.das.mbpast.ReplicaManager#getReplicaSet()
	 */
	@Override
	public NodeHandleSet getReplicaSet() {
		return endpoint.replicaSet(this.node.getId(), alfa);
	}

	@Override
	public boolean forward(RouteMessage message) {
		return true;
	}

	@Override
	public void deliver(Id id, Message message) {
		if (!(message instanceof MBPastMessage)) {
			logger.log("wrong message " + message);
			throw new RuntimeException("Wrong message" + message);
		}
		MBPastMessage mbMessage = (MBPastMessage) message;
		if (mbMessage.getSender().equals(node.getLocalNodeHandle())) {
			logger.log("self message " + message);
			return;
		}
		if (mbMessage instanceof ReplicaTimestampRequest) {
			handleReplicaTimestampRequest((ReplicaTimestampRequest) message);
		} else if (mbMessage instanceof ReplicaTimestampRequestAck) {
			handleReplicaTimestampRequestAck((ReplicaTimestampRequestAck) message);
		} else if (mbMessage instanceof ReplicaIdsMessage) {
			handleReplicaIdsMessage((ReplicaIdsMessage) message);
		} else if (mbMessage instanceof ReplicaRequest) {
			handleReplicaRequest((ReplicaRequest) message);
		} else if (mbMessage instanceof ReplicaRequestAck) {
			handleReplicaRequestAck((ReplicaRequestAck) message);
		} else if (mbMessage instanceof StoreSwarmMessage) {
			handleStoreSwarmMessage((StoreSwarmMessage) message);
		}
		logger.log(message.toString());
	}

	private void handleReplicaTimestampRequestAck(
			ReplicaTimestampRequestAck message) {
		downloader.receivedTimestamp(message);
	}

	private void handleReplicaTimestampRequest(ReplicaTimestampRequest message) {
		if (mbs.hasReplication(message.getReplicaId())) {
			endpoint.route(
					message.getSender().getId(),
					new ReplicaTimestampRequestAck(this.node
							.getLocalNodeHandle(), message.getReplicaId(), mbs
							.getTimestamp(message.getReplicaId()), message
							.getUniqueId()), message.getSender());
		} else {
			logger.log("no replication of " + message.getReplicaId());
		}
	}

	private void handleStoreSwarmMessage(StoreSwarmMessage message) {
		mbs.storeSwarmView(message.getSwarm());
	}

	private void handleReplicaRequestAck(ReplicaRequestAck message) {
		logger.log("received replica");
		downloader.receivedReplication(message);
		// mbs.storeProfile(message.getReplication().getProfile());
	}

	private void handleReplicaRequest(ReplicaRequest message) {
		if (mbs.hasReplication(message.getProfileId())) {
			endpoint.route(
					message.getSender().getId(),
					new ReplicaRequestAck(this.node.getLocalNodeHandle(), mbs
							.getReplication(message.getProfileId()), message),
					message.getSender());
		} else {
			if (message.getProfileId().equals(this.node.getId())) {
				endpoint.route(message.getSender().getId(),
						new ReplicaRequestAck(this.node.getLocalNodeHandle(),
								mbs.getReplication(message.getProfileId()),
								message), message.getSender());
			}
		}
	}

	private synchronized void handleReplicaIdsMessage(
			ReplicaIdsMessage replicaIds) {
		{
			synchronized (knownIds) {
				knownIds.addAll(replicaIds.getReplicas());
				nodeIdsModified = true;
				knownIds.notifyAll();
			}

			for (Id replicaId : replicaIds.getReplicas()) {
				if (!isReplica(replicaId)) {
					ignoredIds.add(replicaId);
				} else {
					if (!mbs.isConsolidated(replicaId)
							&& !downloader.isDownloading(replicaId)) {
						if (!replicaId.equals(node.getId())) {
							ignoredIds.remove(replicaId);
							downloader.scheduleDownload(replicaId);
						}
					}
				}
			}
		}
	}

	@Override
	public synchronized void update(final NodeHandle aNode, boolean joined) {
		logger.log("node join? " + joined);
		
		if (joined) { // XXX: alter..
			scheduleReplicaIdsSend(aNode);
		}
	}

	private void scheduleReplicaIdsSend(final NodeHandle aNode) {
		new Thread(){

			@Override
			public void run() {
				endpoint.route(aNode.getId(),
						new ReplicaIdsMessage(node.getLocalNodeHandle(),
								knownIds, new GregorianCalendar().getTime()),
						aNode, new DeliveryNotification() {
							@Override
							public void sent(MessageReceipt msg) {
								logger.log("sucess sending replicasIds to "
										+ aNode.getId());
							}

							@Override
							public void sendFailed(MessageReceipt msg,
									Exception reason) {
								logger.log("fail to send to " + aNode.getId()
										+ " replicaIds." + " " + reason);
								try {
									env.getTimeSource().sleep(2 * 1000);
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
								logger.log("resending to " + aNode.getId()
										+ " replicaIds.");
									endpoint.route(
											aNode.getId(),
											new ReplicaIdsMessage(node
													.getLocalNodeHandle(), knownIds,
													new GregorianCalendar().getTime()),
													aNode);
							}
						}
				);
			}
			
		}.start();
	}

//	private void sendReplicaIds(NodeHandle handle) {
//		if (isReplica(handle.getId())) {
//			Set<Id> replicasToSend = new HashSet<Id>();
//			for (Id replica : knownIds) {
//				try {
//					if (endpoint.replicaSet(replica, beta).memberHandle(
//							this.node.getId())) {
//						replicasToSend.add(replica);
//					}
//				} catch (Exception e) {
//					logger.logException("Error calculating replicaSet", e);
//				}
//			}
//			if (!replicasToSend.isEmpty()) {
//				endpoint.route(
//						handle.getId(),
//						new ReplicaIdsMessage(node.getLocalNodeHandle(),
//								replicasToSend, new GregorianCalendar()
//										.getTime()), handle);
//			}
//		}
//	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see br.ufsc.das.mbpast.ReplicaManager#getStoredReplicasIds()
	 */
	@Override
	public Set<Id> getKnownIds() {
		synchronized (idsSync) {
			return knownIds;
		}
	}

	@Override
	public Set<Id> getKnownIdsBlocking() {
		synchronized (idsSync) {
			try {
				if (nodeIdsModified) {
					nodeIdsModified = false;
				}
				else {
					idsSync.wait();
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return knownIds;
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see br.ufsc.das.mbpast.ReplicaManager#printReplicas()
	 */
	@Override
	public void printReplicas() {
		System.out.print("replicas stored on : " + this.node.getId() + "[ ");
		for (Id id : getKnownIds()) {
			System.out.print(id + " ");
		}
		System.out.println("]");
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see br.ufsc.das.mbpast.ReplicaManager#replicates(rice.p2p.commonapi.Id)
	 */
	@Override
	public boolean replicates(Id id) {
		return this.getKnownIds().contains(id);
	}

	class IDProximity implements Comparable<IDProximity> {
		private final Id myId;
		private Distance distance;

		public IDProximity(Id id) {
			this.myId = id;
			this.distance = id.distanceFromId(MBPastImpl.this.node.getId());
		}

		public Id getId() {
			return this.myId;
		}

		@Override
		public int compareTo(IDProximity other) {
			return this.distance.compareTo(other.distance);
		}

	}

	@Override
	public void saveTweet(TweetMessage tweet) {
		mbs.storeUpdate(tweet);
	}

	@Override
	public void storeSwarm(Swarm swarm) {
		if (isManager(swarm.getOwner())) {
			mbs.storeSwarmView(swarm);
			NodeHandleSet replicas = getReplicaSet();
			for (int replicaNumber = 0; replicaNumber < replicas.size(); replicaNumber++) {
				endpoint.route(
						replicas.getHandle(replicaNumber).getId(),
						new StoreSwarmMessage(node.getLocalNodeHandle(), swarm),
						replicas.getHandle(replicaNumber));
			}
		} else {
			logger.log("not the manager to store the swarm");
		}
	}

	public Swarm getSwarmView(Id group) {
		if (mbs.hasReplication(group))
			return mbs.getReplication(group).getSwarm();
		return new Swarm(group, env);
	}

}